package com.shux.kafka.consumer;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.Deserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * 描述：
 * 
 * <pre>HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    2017年9月18日      Simba.Hua         Create
 * ****************************************************************************
 * </pre>
 * @author Simba.Hua
 */
public abstract  class ShuxKafkaConsumer<T> {
	private ConsumerConfig consumerConfig;
	private BlockingQueue<KafkaMessage<T>> fetchMessageQueue;
	private BlockingQueue<KafkaBatchMessage<T>> packMessageQueue;
	private FetchMessageThread<T> fetchThread;
	private PackMessageThread<T> packMessage;
	private ConsumerOptionConfig optionConfig;
	private ConsumerMessageSender<T> messageSender;
	private IMessageListener<T> messageListener;
	protected Deserializer<T> deserializer;
	protected KafkaConsumer<String, T> consumer;
	protected Logger logger = LoggerFactory.getLogger(getClass());
	
	public ShuxKafkaConsumer(ConsumerConfig consumerConfig,ConsumerOptionConfig optionConfig,IMessageListener<T> messageListener,Deserializer<T> deserializer){
		this.consumerConfig = consumerConfig;
		if(optionConfig == null) optionConfig = new ConsumerOptionConfig();
		this.optionConfig = optionConfig;
		this.messageListener = messageListener;
		this.deserializer = deserializer;
		fetchMessageQueue = new ArrayBlockingQueue<>(optionConfig.getFatchQueueSize());
		packMessageQueue = new ArrayBlockingQueue<>(optionConfig.getPackDataQueueSize());
		initConnector();
		initFetchMessage();
		initPackMessage();
		sendConsumerMessage();
		commitOffsetToKafka();
	}
	/**
	 * 从服务端producer抓取数据
	 */
	private void initFetchMessage(){
		fetchThread = new FetchMessageThread<>(null,fetchMessageQueue,consumer);
		fetchThread.start();
	}
	/**
	 * 把从producer的数据封装成批量的数据，以便批量推送给客户端consumer
	 */
	private void initPackMessage() {
		packMessage = new PackMessageThread<>(null,fetchMessageQueue,packMessageQueue,optionConfig,this);
		packMessage.start();
	}
	/**
	 * 多线程发送给客户端consumer
	 */
	private void sendConsumerMessage() {
		messageSender = new ConsumerMessageSender<>(optionConfig,packMessageQueue,this,messageListener);
		messageSender.sendConsumerMessage();
	}
	
	protected void commitOffsetToKafka(){
		
	}
	protected void completeMessage(KafkaBatchMessage<T> kafkaBatchMessage) {};
	//在把数据加入到PackMessagQueue前先校验数据是否发送过，以免重复发送，在不可重复发送的情况下需要实现这个方法
	protected abstract void beforeAddPackMessageToQueue(KafkaBatchMessage<T> batchMessage);
	
	private void initConnector() {
		KafkaConsumer<String, T> consumer = new KafkaConsumer<>(initOptionConfig());
		consumer.subscribe(Arrays.asList(consumerConfig.getTopic()));
		this.consumer = consumer;
	}
	
	private Properties initOptionConfig() {
		Properties config = new Properties();
		config.setProperty("group.id", consumerConfig.getSystemIdToken());
		config.setProperty("bootstrap.servers", consumerConfig.getKafkaServer());
		config.setProperty("key.deserializer", deserializer.getClass().getName());
		config.setProperty("value.deserializer", deserializer.getClass().getName());
		config.setProperty("enable.auto.commit", String.valueOf(optionConfig.getTransation()));
		config.setProperty("auto.commit.interval.ms", optionConfig.getAutoCommitIntervalMs().toString());
		config.setProperty("session.timeout.ms", optionConfig.getSessionTimeout().toString());
		config.setProperty("zookeeper.session.timeout.ms", optionConfig.getZookeeperSessionTimout().toString());
		config.setProperty("zookeeper.connection.timeout.ms", optionConfig.getZookeeperConnectionTimeout().toString());
		config.setProperty("fetch.max.bytes", optionConfig.getFetchMessageMaxBytes().toString());
		config.setProperty("auto.offset.reset", optionConfig.getAutoOffsetReset());
		config.setProperty("auto.offset.reset.ms", optionConfig.getAutoOffetResetMs().toString());
		config.setProperty("rebalance.max.retries", optionConfig.getRebalanceMaxRetries().toString());
		
		return config;
	}
	public void close(){
		fetchThread.shutdown();
		packMessage.shutdown();
		messageSender.getExecutorService().shutdownNow();
		consumer.close();
	}
	public ConsumerConfig getConsumerConfig() {
		return consumerConfig;
	}
	
	
}
